package avicit.bdp.dcs.semistructure.base;


import avicit.bdp.dcs.detail.dto.DataDetailDto;
import avicit.bdp.dcs.detail.dto.TableDetailDto;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.semistructure.dto.HiveJsonDto;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.threshold.utils.TestUtils;
import avicit.bdp.dcs.utils.DcsConstants;
import avicit.platform6.core.rest.msg.ResponseMsg;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.springframework.boot.web.client.RestTemplateBuilder;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.client.RestTemplate;

import java.util.Base64;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

import static org.junit.Assert.assertEquals;

public class NewHdfs2HiveTest {

    public static final String USER_NAME = "avicit2015";
    public static final String PASS_WORD = "avicit2015";
    public static final String URL_PREFIX = "http://127.0.0.1:20002";
    private String jobName = "半结构数据同步hdfs2hive门槛用例";
    private String dstTableName = "liyb_1204";
    private String templateId = "4028809b76263b3e017626601d3c0010";
    private String filePath = "hdfs://192.168.0.28:8020/dev/dcs/chenlei/bcd/ddd";
    private int skipLines = 1;

    /**
     * 所有用例执行前准备工作
     */
    @BeforeClass
    public static void init() {
        System.out.println("Hdfs2Hive门槛用例用例初始化...");
    }

    /**
     * 单个测试用例运行前准备工作
     */
    @Before
    public void setUp() throws Exception{
    }

    /**
     * 单个测试用例运行后销毁工作
     */
    @After
    public void tearDown() {
    }

    /**
     * 所有用例执行完成后准备工作
     */
    @AfterClass
    public static void destory() {
        System.out.println("Hdfs2Hive门槛用例执行完成，开始清理...");
    }

    /***********************************************Begin 测试用例 ***********************************************/
    /**
     * 测试txt设备文件从hdfs到hive
     * @throws Exception
     */
    @Test(timeout = 5 * 60 * 1000)
    public void testHdfs2Hive() throws Exception {
        RestTemplate restTemplate  = new RestTemplateBuilder().basicAuthentication(USER_NAME, PASS_WORD).build();
        String jobId = "";
        try{
            // 1：保存数据源
            ResponseEntity<ResponseMsg> saveDataSourceResult = restTemplate.postForEntity(URL_PREFIX
                            + "/api/bdp/dcs/job/JobRest/saveDataSource", TestUtils.buildSaveDataSourceParas(3,
                    jobName, TestUtils.HDFS_DATASOURCE_ID, TestUtils.HIVE_DATASOURCE_ID, 1),
                    ResponseMsg.class);
            assertEquals("200", saveDataSourceResult.getBody().getRetCode());
            System.out.println("1：保存数据源成功");

            // 获取创建的JobId
            ResponseMsg responseMsg = saveDataSourceResult.getBody();
            Map<String, String> jobMap = (Map<String, String>)responseMsg.getResponseBody();
            jobId = jobMap.get("jobId");

            // 2：保存文件和hive表映射关系,hive表名引用文件名
            ResponseEntity<ResponseMsg> result = restTemplate.postForEntity(URL_PREFIX + "/api/bdp/dcs/job/JobRest/saveTableMapping",
                    buildSaveTblMapping(3, jobId, templateId, dstTableName, filePath), ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());
            System.out.println("2：保存文件和hive表映射关系成功");

            // 3：获取任务信息
            result = restTemplate.getForEntity(URL_PREFIX + "/api/bdp/dcs/task/TaskRest/getTaskById?jobId=" +
                            jobId + "&srcTableName=" + templateId,
                    ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());
            System.out.println("3：获取任务信息成功");
            Map taskMap = (Map) result.getBody().getResponseBody();
            // 获取任务id
            String taskId = (String) taskMap.get("id");

            // 4：修改任务配置
            result = restTemplate.postForEntity(URL_PREFIX + "/api/bdp/dcs/task/TaskRest/updateTask",
                    buildUpdateTaskParam(taskId), ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());
            System.out.println("4：修改任务配置成功");

            // 5：任务配置检查
            result = restTemplate.postForEntity(URL_PREFIX + "/api/bdp/dcs/job/JobRest/checkTaskConfig",
                    TestUtils.buildCheckTaskConfig(jobId), ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());
            System.out.println("5：任务配置检查成功");

            // 6：预检查
            result = restTemplate.postForEntity(URL_PREFIX + "/api/bdp/dcs/precheck/PrecheckRest" +
                            "/executePrecheck",
                    TestUtils.buildPreChecks(jobId), ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());
            System.out.println("6：预检查成功");

            // 7：更新Job状态
            result = restTemplate.postForEntity(URL_PREFIX + "/api/bdp/dcs/job/JobRest/updateJobStatus",
                    TestUtils.buildUpdateJobStatus(jobId), ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());
            System.out.println("7：更新Job状态成功");

            // 8：启动任务
            result = restTemplate.postForEntity(URL_PREFIX + "/api/bdp/dcs/job/JobRest/startJobByIds",
                    TestUtils.buildStartJob(jobId), ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());
            System.out.println("8：启动任务成功");

            // 获取Job列表
            // 等待DS启动任务成功
            Thread.sleep(50 * 1000);
            result = restTemplate.exchange(URL_PREFIX + "/api/bdp/dcs/job/JobRest/getJobList" +
                            "?jobStatus&sort&name=" + jobName + "&jobType=3&currentPage=1&pageSize=10",
                    HttpMethod.GET, TestUtils.buildGetHttpEntity(), ResponseMsg.class);
            ResponseMsg jobListBody = result.getBody();
            LinkedHashMap<String, Object> map = (LinkedHashMap<String, Object>)jobListBody.getResponseBody();
            List<Object> finalBody = (List<Object>) map.get("result");
            for (Object obj : finalBody) {
                LinkedHashMap<String, String> jobObj = (LinkedHashMap<String, String>)obj;
                JobDTO jobInfo = JSON.parseObject(JSON.toJSONString(jobObj), JobDTO.class);
                if (jobInfo.getId().equals(jobId)) {
                    System.out.println("半结构数据同步门槛用例任务状态:" + jobInfo.getJobStatus());

                    // 9：判断任务是否同步完成;
                    assertEquals(7 == jobInfo.getJobStatus(), true);
                    System.out.println("9：判断任务是否同步完成成功");

//                    // 17：判断同步数据大小;（备注：源表中只有一行数据，大小固定）
//                    assertEquals(jobInfo.getReadSucceedBytes() == 8670, true);
//                    System.out.println("17：判断同步数据大小成功");
                    break;
                }
            }

            // 10：查看日志信息
            result = restTemplate.exchange(URL_PREFIX + "/api/bdp/dcs/log/LogRest/getLogByTaskId" +
                    "?taskInstanceId=&skipLineNum=0" +
                    "&limit=100&jobId="+jobId, HttpMethod.GET, TestUtils.buildGetHttpEntity(), ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());
            System.out.println("10：查看日志信息成功");

            // 11：查看详情，表结构迁移；
            result = restTemplate.exchange(URL_PREFIX + "/api/bdp/dcs/detail/TaskTableDetailRest" +
                            "/getTaskTableDetailList" +
                            "?currentPage=1&pageSize=20&srcTableName=&jobId="+jobId, HttpMethod.GET,
                    TestUtils.buildGetHttpEntity(), ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());
            System.out.println("11：查看详情，表结构迁移成功");
            ResponseMsg body = result.getBody();
            map = (LinkedHashMap<String, Object>)body.getResponseBody();
            finalBody = (List<Object>) map.get("result");
            for (Object obj : finalBody) {
                LinkedHashMap<String, String> jobObj = (LinkedHashMap<String, String>)obj;
                TableDetailDto tableDetailDto = JSON.parseObject(JSON.toJSONString(jobObj), TableDetailDto.class);
                // 判断任务task是否同步遷移完成;
                assertEquals(7 == tableDetailDto.getTaskStatus(), true);
                System.out.println("11.1：判断任务task是否同步遷移完成成功");
            }

            // 12：查看详情，数据迁移
            result = restTemplate.exchange(URL_PREFIX + "/api/bdp/dcs/detail/TaskDataDetailRest" +
                            "/getTaskDataDetailList" +
                            "?currentPage=1&pageSize=20&srcTableName=&jobId="+jobId, HttpMethod.GET,
                    TestUtils.buildGetHttpEntity(),
                    ResponseMsg.class);
            assertEquals("200", result.getBody().getRetCode());
            System.out.println("12：查看详情，数据迁移成功");
            body = result.getBody();
            map = (LinkedHashMap<String, Object>)body.getResponseBody();
            finalBody = (List<Object>) map.get("result");
            for (Object obj : finalBody) {
                LinkedHashMap<String, String> jobObj = (LinkedHashMap<String, String>)obj;
                DataDetailDto dataDetailDto = JSON.parseObject(JSON.toJSONString(jobObj), DataDetailDto.class);

                // 判断任务是否task是否同步遷移完成
                assertEquals(7 == dataDetailDto.getTaskStatus(), true);
                System.out.println("12.1：判断任务是否task是否同步遷移完成成功");

                // 判斷完成總數是否為1000條
//                assertEquals(dataDetailDto.getEstimateTotal() == 2, true);

                // 判斷已完成總數是否為1000條
                assertEquals(dataDetailDto.getCompletedTotal() == 1001 - skipLines, true);
                System.out.println("12.2：判斷已完成總數是否為1000條成功");
            }
        } finally {
            // 13：删除job
//            ResponseEntity<ResponseMsg> deleteJobResult = restTemplate.exchange(URL_PREFIX + "/api/bdp/dcs" +
//                            "/job/JobRest" +
//                            "/deleteJobByIds",
//                    HttpMethod.DELETE,
//                    TestUtils.buildDeleteJob(jobId),
//                    ResponseMsg.class);
//            assertEquals("200", deleteJobResult.getBody().getRetCode());
            System.out.println("13：删除job成功");
        }
    }

    private HttpEntity<String> buildUpdateTaskParam(String taskId) {
        TaskDTO taskDTO = new TaskDTO();
        taskDTO.setId(taskId);
        HiveJsonDto hiveJsonDto = new HiveJsonDto();

        hiveJsonDto.setDstTable(dstTableName);
        hiveJsonDto.setEmptyLineProcess(1);
        hiveJsonDto.setFileName(dstTableName);
        hiveJsonDto.setFileSeparator(",");
        hiveJsonDto.setFileType(3);
        hiveJsonDto.setPath(DcsConstants.HIVE_TABLE_LOCATION + dstTableName);
        hiveJsonDto.setSkipFileHeader(1);
        hiveJsonDto.setSkipLine(skipLines);
        hiveJsonDto.setWriteMode(1);
        taskDTO.setHiveJsonDto(hiveJsonDto);
        taskDTO.setModifyColumn(0);
        taskDTO.setMappingColumn("[{\"desensitizationPolicyId\":\"\",\"desensitizationRuleId\":\"\",\"dstColumnName\":\"name\",\"dstColumnType\":\"TEXT\",\"srcColumnName\":\"name\",\"srcColumnType\":\"TEXT\"},{\"desensitizationPolicyId\":\"\",\"desensitizationRuleId\":\"\",\"dstColumnName\":\"age\",\"dstColumnType\":\"TEXT\",\"srcColumnName\":\"age\",\"srcColumnType\":\"TEXT\"},{\"desensitizationPolicyId\":\"\",\"desensitizationRuleId\":\"\",\"dstColumnName\":\"work\",\"dstColumnType\":\"TEXT\",\"srcColumnName\":\"work\",\"srcColumnType\":\"TEXT\"}]");
        String param = JSONObject.toJSONString(taskDTO);
        HttpHeaders headers = new HttpHeaders(); // 创建一个头部对象
        headers.setContentType(MediaType.valueOf("application/json; charset=UTF-8"));
        headers.add("Accept", MediaType.APPLICATION_JSON_UTF8.toString());
        headers.add ("Authorization", "Basic " +
                Base64.getUrlEncoder().encodeToString((USER_NAME + ":" + PASS_WORD).getBytes()));
        return new HttpEntity<String>(param, headers);
    }

    /**
     * 保存表映射关系参数
     */
    public static HttpEntity<String> buildSaveTblMapping(int jobType, String jobId,
                                                         String srcTblName, String dstTbleName, String filePath) {
        JobDTO jobDTO = new JobDTO();
        jobDTO.setId(jobId);
        jobDTO.setTransferType("1,2");
        jobDTO.setProcessMode(1);
        jobDTO.setSelectedTable("[{\"src_table\":\"" + srcTblName+ "\",\"dst_table\":\"" + dstTbleName + "\"}]");
        jobDTO.setSelectedFile(filePath);
//        jobDTO.setExtraParams(dstTbleName);

        // 数据同步
        if (jobType == 1) {
            jobDTO.setJobType(1);
        }
        // 数据同步-增量同步
        else if (jobType == 2) {
            jobDTO.setSyncInterval(1800);
            jobDTO.setTableInit(1);
            jobDTO.setJobType(2);
        } else if (jobType == 3) {
            jobDTO.setSyncInterval(60);
            jobDTO.setTableInit(1);
            jobDTO.setJobType(3);
        }

        String param = JSONObject.toJSONString(jobDTO);
        HttpHeaders headers = new HttpHeaders(); // 创建一个头部对象
        headers.setContentType(MediaType.valueOf("application/json; charset=UTF-8"));
        headers.add("Accept", MediaType.APPLICATION_JSON_UTF8.toString());
        headers.add ("Authorization", "Basic " +
                Base64.getUrlEncoder().encodeToString((USER_NAME + ":" + PASS_WORD).getBytes()));
        return new HttpEntity<String>(param, headers);
    }

}
